package flowprocess

import (
	"context"
	"sync"
)

type NodeKey string

const (
	NODE_ID NodeKey = "nodeId"
)

type node struct {
	processors              []Processor
	processorChanBufferSize int
	done                    chan struct{}
	id                      int
}

func NewNode(processors []Processor, processorChanBufferSize int, id int) Node {
	pu := &node{
		processors:              processors,
		processorChanBufferSize: processorChanBufferSize,
		id:                      id,
	}
	return pu
}

func (pu *node) Process(in TaskChan) (out TaskChan) {
	pu.done = make(chan struct{})

	outchain := make(TaskChan, pu.processorChanBufferSize)

	var wg sync.WaitGroup
	wg.Add(len(pu.processors))

	ctx, cancelFunc := context.WithCancel(context.WithValue(context.Background(), NODE_ID, pu.id))
	for i, processor := range pu.processors {
		go func(index int, processor Processor, inTaskChan <-chan Task, outTaskChan chan<- Task) {
			cancelAllProcess := processor.Proccess(inTaskChan, outTaskChan, ctx)
			if cancelAllProcess {
				cancelFunc()
			}
			wg.Done()
		}(i, processor, in, outchain)
	}

	go func() {
		wg.Wait()
		cancelFunc()
		close(outchain)
		close(pu.done)
	}()

	return outchain
}

func (pu *node) Done() <-chan struct{} {
	return pu.done
}
